/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.common.task.engine;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.common.executor.ExecutorFactory;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.task.AbstractDelayTask;
import com.alibaba.nacos.common.task.NacosTaskProcessor;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Nacos delay task execute engine.
 *
 * @author xiweng.yy
 */
public class NacosDelayTaskExecuteEngine extends AbstractNacosTaskExecuteEngine<AbstractDelayTask> {

    private final ScheduledExecutorService processingExecutor;

    /**
     * 任务缓存
     */
    protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;

    protected final ReentrantLock lock = new ReentrantLock();

        public NacosDelayTaskExecuteEngine(String name) {
        this(name, null);
    }

    public NacosDelayTaskExecuteEngine(String name, Logger logger) {
        this(name, 32, logger, 100L);
    }

    /**
     * 用单一的线程池来启动，然后处理任务
     *
     * @param name
     * @param initCapacity
     * @param logger
     * @param processInterval
     */
    public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
        super(logger);
        tasks = new ConcurrentHashMap<>(initCapacity);
        // 线程池
        processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));

        // 最后调用自己的方法来处理任务
        processingExecutor
                .scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
    }

    @Override
    public int size() {
        lock.lock();
        try {
            return tasks.size();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean isEmpty() {
        lock.lock();
        try {
            return tasks.isEmpty();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 任务满足运行条件，就从任务缓存中移除
     *
     * @param key key of task
     * @return
     */
    @Override
    public AbstractDelayTask removeTask(Object key) {
        lock.lock();
        try {
            AbstractDelayTask task = tasks.get(key);
            if (null != task && task.shouldProcess()) {
                return tasks.remove(key);
            } else {
                return null;
            }
        } finally {
            lock.unlock();
        }
    }

    @Override
    public Collection<Object> getAllTaskKeys() {
        Collection<Object> keys = new HashSet<>();
        lock.lock();
        try {
            keys.addAll(tasks.keySet());
        } finally {
            lock.unlock();
        }
        return keys;
    }

    @Override
    public void shutdown() throws NacosException {
        tasks.clear();
        processingExecutor.shutdown();
    }

    /**
     * 添加任务
     *
     * @param key  key of task
     * @param newTask task
     */
    @Override
    public void addTask(Object key, AbstractDelayTask newTask) {
        lock.lock();
        try {
            // 如果任务已存在，
            AbstractDelayTask existTask = tasks.get(key);
            if (null != existTask) {
                // 那么将任务进行合并
                newTask.merge(existTask);
            }

            // 向任务缓存中插入数据
            tasks.put(key, newTask);
        } finally {
            lock.unlock();
        }
    }

    /**
     * process tasks in execute engine.
     */
    protected void processTasks() {
        /**
         * 获取所有的 key，因为 {@link #addTask(Object, AbstractDelayTask)} 方法，所以每一个 task 都会关联到一个 key
         */
        Collection<Object> keys = getAllTaskKeys();

        // 对所有的任务进行遍历
        for (Object taskKey : keys) {

            // 判断 task 是否到期，每个 task 创建时都是指定 taskInterval
            AbstractDelayTask task = removeTask(taskKey);
            if (null == task) {
                continue;
            }

            /**
             * 获取相应的 processor 来处理，一般来说就是默认的 processor,
             * 这个值是 {@link com.alibaba.nacos.naming.push.v2.task.PushDelayTaskExecuteEngine#PushDelayTaskExecuteEngine(com.alibaba.nacos.naming.core.v2.client.manager.ClientManager, com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager, com.alibaba.nacos.naming.core.v2.index.ServiceStorage, com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager, com.alibaba.nacos.naming.push.v2.executor.PushExecutor, com.alibaba.nacos.naming.misc.SwitchDomain)}
             * 中的初始化方法 中的 setDefaultTaskProcessor 设置的值
             */
            NacosTaskProcessor processor = getProcessor(taskKey);


            try {
                // ReAdd task if process failed
                // 处理 task，如果处理失败，重新添加 task
                if (!processor.process(task)) {
                    // 重新添加 task
                    retryFailedTask(taskKey, task);
                }
            } catch (Throwable e) {
                getEngineLog().error("Nacos task execute error ", e);
                retryFailedTask(taskKey, task);
            }
        }
    }

    /**
     * 任务执行失败的时候， 重新添加 task
     *
     * @param key
     * @param task
     */
    private void retryFailedTask(Object key, AbstractDelayTask task) {
        task.setLastProcessTime(System.currentTimeMillis());
        addTask(key, task);
    }


    /**
     * 处理任务
     * 一个内部类用于父类来调用
     */
    private class ProcessRunnable implements Runnable {
        
        @Override
        public void run() {
            try {
                /**
                 * 这里又会调用到父类 {@link NacosDelayTaskExecuteEngine#processTasks()}
                 */
                processTasks();
            } catch (Throwable e) {
                getEngineLog().error(e.toString(), e);
            }
        }
    }
}
